package com.gome.boot.config.mq.rocketmq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.*;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.gome.rocketmq.client.extension.MessageConsumer;
import com.gome.rocketmq.client.extension.MessageExtConsumer;
import com.gome.rocketmq.client.extension.impl.MessagePushConsumerImpl;
import com.gome.utils.DiamondUtil;
import com.taobao.diamond.manager.ManagerListenerAdapter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;

/**
 * 扩展消息消费扩展接口.
 *
 * @author baoxiufeng
 */
public class MessagePushConsumerExtImpl extends MessagePushConsumerImpl {
    private static final Logger logger = LoggerFactory.getLogger(MessagePushConsumerExtImpl.class);
    private SerializeType serializeType;
    private String consumerGroup;
    private String topic;
    private String subExpression;
    private String messageModel;
    private Integer consumeThreadMin;
    private Integer consumeThreadMax;
    private Integer consumeMessageBatchMaxSize;
    private DefaultMQPushConsumer consumer;
    private String dataId;
    private String group;
    private String namesrvAddr;
    private String instanceName;
    private List<MessageConsumer> messageConsumers;
    private boolean orderly;

    public MessagePushConsumerExtImpl() {
        this.serializeType = SerializeType.FASTJSON;
        this.orderly = false;
    }

    @Override
    public void destroy() throws Exception {
        this.consumer.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.registDiamondListener();
    }

    /**
     * 注册Diamond监听器.
     */
    private void registDiamondListener() {
        DiamondUtil.registDiamondListener(new ManagerListenerAdapter() {
            @Override
            public void receiveConfigInfo(String configInfo) {
                MessagePushConsumerExtImpl.logger.info("group=" + MessagePushConsumerExtImpl.this.group + ",dataId=" + MessagePushConsumerExtImpl.this.dataId + "==>receiveConfigInfo=" + configInfo);
                MessagePushConsumerExtImpl.this.restartConsumer(configInfo);
            }
        }, this.group, this.dataId);
    }

    @Override
    public void restartConsumer() {
        if (this.namesrvAddr == null && "".equals(this.namesrvAddr)) {
            throw new IllegalStateException("namesrvAddr is empty! ");
        } else {
            this.restartConsumer(this.namesrvAddr);
        }
    }

    @Override
    public void restartConsumer(String configInfo) {
        this.namesrvAddr = configInfo;
        DefaultMQPushConsumer consumerNew = new DefaultMQPushConsumer(StringUtils.defaultIfBlank(this.consumerGroup, this.topic));
        if (this.instanceName != null) {
            consumerNew.setInstanceName(this.instanceName);
        }
        consumerNew.setNamesrvAddr(configInfo);
        try {
            consumerNew.subscribe(this.topic, this.subExpression);
            if (this.messageModel != null) {
                consumerNew.setMessageModel(MessageModel.valueOf(this.messageModel));
            }
            if (this.consumeThreadMin != null) {
                consumerNew.setConsumeThreadMin(this.consumeThreadMin.intValue());
            }
            if (this.consumeThreadMax != null) {
                consumerNew.setConsumeThreadMax(this.consumeThreadMax.intValue());
            }
            if (this.consumeMessageBatchMaxSize != null) {
                consumerNew.setConsumeMessageBatchMaxSize(this.consumeMessageBatchMaxSize.intValue());
            }
            consumerNew.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            if (this.orderly) {
                consumerNew.registerMessageListener(new MessageListenerOrderly() {
                    @Override
                    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                        return consumerMessage(msgs, ConsumeOrderlyStatus.SUCCESS, ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT);
                    }
                });
            } else {
                consumerNew.registerMessageListener(new MessageListenerConcurrently() {
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                        return consumerMessage(msgs, ConsumeConcurrentlyStatus.CONSUME_SUCCESS, ConsumeConcurrentlyStatus.RECONSUME_LATER);
                    }
                });
            }
            consumerNew.start();
            logger.info(" DefaultMQPushConsumer Started.");
            DefaultMQPushConsumer consumerOld = this.consumer;
            this.consumer = consumerNew;
            if (consumerOld != null) {
                consumerOld.shutdown();
            }
        } catch (Exception e) {
            logger.error("rocketmq restartConsumer failure.", e);
        }
    }

    /**
     * 消费消息处理.
     *
     * @param msgs 消息列表
     * @param successStatus 消费成功状态
     * @param failureStatus 消费失败状态
     * @param <T> 消息结果状态类型泛型
     * @return 消费结果状态
     */
    private <T> T consumerMessage(List<MessageExt> msgs, T successStatus, T failureStatus) {
        Iterator<MessageExt> msgIter = msgs.iterator();
        while (msgIter.hasNext()) {
            MessageExt msg = msgIter.next();
            Object ms = MessagePushConsumerExtImpl.this.deserializeMsg(msg.getBody());
            Iterator<MessageConsumer> consumerIter = MessagePushConsumerExtImpl.this.messageConsumers.iterator();
            while (consumerIter.hasNext()) {
                MessageConsumer consumer = consumerIter.next();
                try {
                    MessagePushConsumerExtImpl.logger.info("rocketmq consumer instance={}, receive message:{}", consumer, msg);
                    if (consumer instanceof MessageExtConsumer) {
                        ((MessageExtConsumer) consumer).consume(msg, ms);
                    } else {
                        RocketMqUtil.getUtil().setMsgId(msg.getMsgId());
                        consumer.consume(ms);
                    }
                } catch (ClassCastException e) {
                    // 如果是参数转换异常，则只打印异常信息，不再重试消费
                    MessagePushConsumerExtImpl.logger.error("rocketmq consumer instance=" + consumer + ",receive message:" + msg, e);
                } catch (Exception e) {
                    MessagePushConsumerExtImpl.logger.error("rocketmq consumer msgId=" + msg.getMsgId(), e);
                    return failureStatus;
                }
            }
        }
        return successStatus;
    }

    @Override
    protected Object deserializeMsg(byte[] body) {
        return this.serializeType.deserialize(body);
    }

    @Override
    public void decThreadNum(int corePoolSize) {
        this.consumer.getDefaultMQPushConsumerImpl().getConsumeMessageService().updateCorePoolSize(corePoolSize);
    }

    @Override
    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    @Override
    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    public void setSubExpression(String subExpression) {
        this.subExpression = subExpression;
    }

    @Override
    public void setMessageConsumers(List<MessageConsumer> messageConsumers) {
        this.messageConsumers = messageConsumers;
    }

    @Override
    public String getMessageModel() {
        return this.messageModel;
    }

    @Override
    public void setMessageModel(String messageModel) {
        this.messageModel = messageModel;
    }

    @Override
    public void setConsumeThreadMax(Integer consumeThreadMax) {
        this.consumeThreadMax = consumeThreadMax;
    }

    @Override
    public void setConsumeThreadMin(Integer consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    @Override
    public void setConsumeMessageBatchMaxSize(Integer consumeMessageBatchMaxSize) {
        this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
    }

    @Override
    public void setDataId(String dataId) {
        this.dataId = dataId;
    }

    @Override
    public void setGroup(String group) {
        this.group = group;
    }

    @Override
    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    @Override
    public String getNamesrvAddr() {
        return this.namesrvAddr;
    }

    @Override
    public String getInstanceName() {
        return this.instanceName;
    }

    @Override
    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    @Override
    public void setDiamondIpList(String diamondIpList) {
        logger.warn("不推荐使用这行配置,建议配置diamond配置文件ServerAddress：<property name=\"diamondIpList\" value=\"${diamond.ipList}\" />");
    }

    public void setSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
    }

    public MessagePushConsumerExtImpl resetSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
        return this;
    }

    @Override
    public boolean isOrderly() {
        return this.orderly;
    }

    @Override
    public void setOrderly(boolean orderly) {
        this.orderly = orderly;
    }
}
